package db.freebase;

import static db.freebase.FreeBaseUtils.isValue;
import static db.freebase.FreeBaseUtils.midToInt;
import static db.freebase.FreeBaseUtils.parseLine;
import static common.utils.CommonUtils.wrapStringWithQuotes;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import common.types.Pair;

import db.freebase.entities.FreeBaseFictionalCharacter;
import db.freebase.entities.FreeBaseFictionalDateTime;
import db.freebase.entities.FreeBaseFictionalEmpolymentTenure;
import db.freebase.entities.FreeBaseGenericMediator;
import db.manager.DBManager;

public class FreeBaseDBImporter {

    private static final int BATCH_SIZE = 10000;
    private static final int LIST_SIZE = 1000;
    
    private FreeBaseDBImporter() { }
    
    public static void importToDB(File freeBaseRoot) throws IOException, SQLException {
        
        File freeBaseFictionalCharacters = loadFile(freeBaseRoot, "fictional_character");
        File freeBaseEmploymentTenure = loadFile(freeBaseRoot, "fictional_employment_tenure");
        File freeBaseFictionalTimeDate = loadFile(freeBaseRoot, "fictional_date_time");
        File freeBaseMarriage = loadFile(freeBaseRoot, "marriage_of_fictional_characters");
        File freeBaseRomantic = loadFile(freeBaseRoot, "romantic_involvement");
        File freeBaseSiblings = loadFile(freeBaseRoot, "sibling_relationship_of_fictional_characters");

        BatchExecutor executor = new BatchExecutor();
        
        FreeBaseTSVReader reader = null;
        FreeBaseTSVLine line;
        
        try {
        
            executor.addBatch(createTempEmploymentTenureTableStatement());
            executor.addBatch(createTempDateTimeTableStatement());
            executor.addBatch(createTempSibligsTableStatement());
            executor.addBatch(createTempMarriedTableStatement());
            executor.addBatch(createTempRomanticTableStatement());
            executor.forceBatch();
            
            // ####################################################################
            
            reader = new FreeBaseTSVReader(freeBaseEmploymentTenure.getAbsolutePath());
            while ((line = reader.readLine()) != null) {
                FreeBaseFictionalEmpolymentTenure tenure = parseLine(line, FreeBaseFictionalEmpolymentTenure.class);
                if (isValue(tenure.getMid()) && isValue(tenure.getEmployer()) && isValue(tenure.getEmployee())) {
                    int mid = midToInt(tenure.getMid());
                    executor.addBatch(propertyInsertStatement("Employer", new String[] { tenure.getEmployer() }));
                    executor.addBatch(employmentTenureInsertStatement(mid, tenure.getTitle(), tenure.getEmployer()));
                }
            }
            executor.forceBatch();
            reader.close();
            
            // ####################################################################
            
            reader = new FreeBaseTSVReader(freeBaseFictionalTimeDate.getAbsolutePath());
            while ((line = reader.readLine()) != null) {
                FreeBaseFictionalDateTime dateTime = parseLine(line, FreeBaseFictionalDateTime.class);
                if (isValue(dateTime.getMid())) {
                    DateTimeEntry entry = parseDateTimeLine(dateTime);
                    executor.addBatch(dateTimeInsertStatement(entry));
                }
            }
            executor.forceBatch();
            reader.close();
            
            // ####################################################################

            reader = new FreeBaseTSVReader(freeBaseMarriage.getAbsolutePath());
            while ((line = reader.readLine()) != null) {
                FreeBaseGenericMediator marriage = parseLine(line, FreeBaseGenericMediator.class);
                if (isValue(marriage.getMid()) && isValue(marriage.getValues()) &&
                    marriage.getValues().length >= 2) {
                    executor.addBatch(marriageInsertStatement(marriage));
                }
            }
            executor.forceBatch();
            reader.close();
            
            // ####################################################################
            
            reader = new FreeBaseTSVReader(freeBaseRomantic.getAbsolutePath());
            while ((line = reader.readLine()) != null) {
                FreeBaseGenericMediator romantic = parseLine(line, FreeBaseGenericMediator.class);
                if (isValue(romantic.getMid()) && isValue(romantic.getValues()) && 
                    romantic.getValues().length >= 2) {
                    executor.addBatch(romanticInsertStatement(romantic));
                }
            }
            executor.forceBatch();
            reader.close();
            
            // ####################################################################
            
            reader = new FreeBaseTSVReader(freeBaseSiblings.getAbsolutePath());
            while ((line = reader.readLine()) != null) {
                FreeBaseGenericMediator siblings = parseLine(line, FreeBaseGenericMediator.class);
                if (isValue(siblings.getMid()) && isValue(siblings.getValues()) &&
                    siblings.getValues().length >= 2) {
                    executor.addBatch(siblingsInsertStatement(siblings));
                }
            }
            executor.forceBatch();
            reader.close();
            
            // ####################################################################
            
            reader = new FreeBaseTSVReader(freeBaseFictionalCharacters.getAbsolutePath());
            while ((line = reader.readLine()) != null) {
                FreeBaseFictionalCharacter character = parseLine(line, FreeBaseFictionalCharacter.class);
                if (isValue(character.getName()) && isValue(character.getMid())) {
                    executor.addBatch(characterInsertStatment(character));
                    executor.addBatch(propertyInsertStatement("BasedOn", character.getBasedOn()));
                    executor.addBatch(propertyInsertStatement("Quotation", character.getQuotations(), false));
                    executor.addBatch(propertyInsertStatement("Objects", character.getHasPossesed()));
                    executor.addBatch(propertyInsertStatement("Species", character.getSpecies()));
                    executor.addBatch(propertyInsertStatement("Occupation", character.getOccupation()));
                    executor.addBatch(propertyInsertStatement("Ranks", character.getRank()));
                    executor.addBatch(propertyInsertStatement("Organization", character.getOrganizations()));
                    executor.addBatch(propertyInsertStatement("Powers", character.getPowersOrAbilities()));
                    executor.addBatch(propertyInsertStatement("Universe", character.getAppearsInTheseFictinalUniverses()));
                    executor.addBatch(propertyInsertStatement("CreatedBy", character.getCharacterCreatedBy()));
                    executor.addBatch(propertyInsertStatement("Ethnicity", character.getEthnicity()));
                    executor.addBatch(propertyInsertStatement("School", character.getEducation()));
                    executor.addBatch(propertyInsertStatement("MedicalCondition", character.getMedicalConditions()));
                    executor.addBatch(propertyInsertStatement("Places", character.getPlaceOfBirth()));
                    executor.addBatch(propertyInsertStatement("Places", character.getPlacesLived()));
                }
            }
            executor.forceBatch();
            reader.close();
            
            // ####################################################################
            
            reader = new FreeBaseTSVReader(freeBaseFictionalCharacters.getAbsolutePath());
            Map<Integer, FreeBaseFictionalCharacter> characters = new HashMap<Integer,FreeBaseFictionalCharacter>();
            do {
                characters.clear();
                while (characters.size() < LIST_SIZE && (line = reader.readLine()) != null) {
                    FreeBaseFictionalCharacter character = parseLine(line, FreeBaseFictionalCharacter.class);
                    if (isValue(character.getName()) && isValue(character.getMid())) {
                        characters.put(midToInt(character.getMid()), character);
                    }
                }
                if (characters.size() == 0) {
                    continue;
                }
                List<EntryTuple> status = readCharactersStatus(characters);
                for (EntryTuple tuple : status) {
                    FreeBaseFictionalCharacter character = tuple.getCharacter();
                    EntryStatus currentStatus = tuple.getStatus();
                    int cid = tuple.getCid();
                    if (currentStatus == EntryStatus.Modified) {
                        // User has modified this entry, so by design we choose not to modify it
                        continue;
                    } else if (currentStatus == EntryStatus.Unmodified) {
                        // Entry already existed before import but it might have changed.
                        // so by design, we remove previous entries regarding this character from previous import
                        // and start fresh
                        removeCharacterProperties(executor, character, cid);
                    } 
                    int mid = midToInt(character.getMid());
                    executor.addBatch(insertSelectStatement(
                        "BasedOn", "CharacterBasedOn", mid, "basedOnID", "basedOnName", character.getBasedOn()));
                    executor.addBatch(insertSelectStatement(
                        "Quotation", "CharacterQuotation", mid, "quotationID", "quotationName", character.getQuotations()));
                    executor.addBatch(insertSelectStatement(
                        "Objects", "CharacterPossesedObject", mid, "objectID", "objectName", character.getHasPossesed()));
                    executor.addBatch(insertSelectStatement(
                        "Species", "CharacterSpecies", mid, "speciesID", "speciesName", character.getSpecies()));
                    executor.addBatch(insertSelectStatement(
                        "Occupation", "CharacterOccupation", mid, "occupationID", "occupationName", character.getOccupation()));
                    executor.addBatch(insertSelectStatement(
                        "Ranks", "CharacterRank", mid, "rankID", "rankName", character.getRank()));
                    executor.addBatch(insertSelectStatement(
                        "Organization", "CharacterOrganization", mid, "organizationID", "organizationName", character.getOrganizations()));
                    executor.addBatch(insertSelectStatement(
                        "Powers", "CharacterPowers", mid, "powerID", "powerName", character.getPowersOrAbilities()));
                    executor.addBatch(insertSelectStatement(
                        "Universe", "CharacterUniverses", mid, "universeID", "universeName", character.getAppearsInTheseFictinalUniverses()));
                    executor.addBatch(insertSelectStatement(
                        "CreatedBy", "CharacterCreatedBy", mid, "createdByID", "createdByName", character.getCharacterCreatedBy()));
                    executor.addBatch(insertSelectStatement(
                        "Ethnicity", "CharacterEthnicity", mid, "ethnicityID", "ethnicityName", character.getEthnicity()));
                    executor.addBatch(insertSelectStatement(
                        "School", "CharacterSchool", mid, "schoolID", "schoolName", character.getEducation()));                 
                    executor.addBatch(insertSelectStatement(
                        "MedicalCondition", "CharacterMedicalCondition", mid, "medicalConditionID", "medicalConditionName", character.getMedicalConditions()));
                    executor.addBatch(insertSelectStatement(
                        "Places", "CharacterPlaceOfBirth", mid, "placeID", "placeName", character.getPlaceOfBirth()));
                    executor.addBatch(insertSelectStatement(
                        "Places", "CharacterPlacesLived", mid, "placeID", "placeName", character.getPlacesLived()));
                    executor.addBatch(insertSelectStatement("CharacterParent", mid, "parentID", character.getParents()));
                    executor.addBatch(insertSelectStatement("CharacterChildren", mid, "childID", character.getChildren()));
                    executor.addBatch(characterEmployerInsertStatement(mid, character.getEmployers()));
                    executor.addBatch(characterDateTimeInsertStatement(mid, character.getDateOfBirth()));
                    executor.addBatch(characterMarriedToInsertStatement(mid, character.getMarriedTo()));
                    executor.addBatch(characterRomanticInvolvementInsertStatement(mid, character.getRomanticallyInvolvedWith()));
                    executor.addBatch(characterSiblingsInsertStatement(mid, character.getSiblings()));
                }
            } while (line != null);
            executor.forceBatch();
            reader.close();
            
            // ####################################################################
            
            executor.addBatch(dropTempEmploymentTenureTableStatement());
            executor.addBatch(dropTempDateTimeTableStatement());
            executor.addBatch(dropTempSibligsTableStatement());
            executor.addBatch(dropTempMarriedTableStatement());
            executor.addBatch(dropTempRomanticTableStatement());
            executor.addBatch(removeNewlyAddedMarkFromFictionalCharactersStatement());
            executor.forceBatch();
            
        } finally {
            if (reader != null) {
                reader.close();
            }
            if (executor != null) {
                executor.close();
            }
        }
    }
    
    private static void removeCharacterProperties(BatchExecutor executor, FreeBaseFictionalCharacter character,
            int cid) throws SQLException {
        executor.addBatch(updateCharacterSelfPropertiesStatement(character, cid));
        executor.addBatch(removePropertyFromStatement("CharacterBasedOn", cid));
        executor.addBatch(removePropertyFromStatement("CharacterQuotation", cid));
        executor.addBatch(removePropertyFromStatement("CharacterPossesedObject", cid));
        executor.addBatch(removePropertyFromStatement("CharacterSpecies", cid));
        executor.addBatch(removePropertyFromStatement("CharacterOccupation", cid));
        executor.addBatch(removePropertyFromStatement("CharacterRank", cid));
        executor.addBatch(removePropertyFromStatement("CharacterOrganization", cid));
        executor.addBatch(removePropertyFromStatement("CharacterPowers", cid));
        executor.addBatch(removePropertyFromStatement("CharacterUniverses", cid));
        executor.addBatch(removePropertyFromStatement("CharacterCreatedBy", cid));
        executor.addBatch(removePropertyFromStatement("CharacterEthnicity", cid));
        executor.addBatch(removePropertyFromStatement("CharacterSchool", cid));
        executor.addBatch(removePropertyFromStatement("CharacterMedicalCondition", cid));
        executor.addBatch(removePropertyFromStatement("CharacterPlaceOfBirth", cid));
        executor.addBatch(removePropertyFromStatement("CharacterPlacesLived", cid));
        executor.addBatch(removePropertyFromStatement("CharacterParent", cid));
        executor.addBatch(removePropertyFromStatement("CharacterChildren", cid));
        executor.addBatch(removePropertyFromStatement("CharacterEmployer", cid));
        executor.addBatch(removePropertyFromStatement("CharacterDateOfBirth", cid));
        executor.addBatch(removePropertyFromStatement("CharacterMarriedTo", cid));
        executor.addBatch(removePropertyFromStatement("CharacterRomanticallyInvolvedWith", cid));
        executor.addBatch(removePropertyFromStatement("CharacterSibling", cid));
    }

    private static String updateCharacterSelfPropertiesStatement(FreeBaseFictionalCharacter character, int cid) {
        
        String[] genders = character.getGender();
        String gender = null;
        if (isValue(genders)) {
            gender = genders[0];
        }
        
        float height = character.getHeight();
        float weight = character.getWeight();
        
        StringBuilder sb = new StringBuilder();
        sb.append("UPDATE FictionalCharacter SET gender = ");
        sb.append(isValue(gender) ? wrapStringWithQuotes(truncate(gender, Property.SHORT)) : "NULL");
        sb.append(", height = ");
        sb.append(isValue(height) ? wrapStringWithQuotes(convertFreeBaseHeightToEnum(height)) : "NULL"); 
        sb.append(", weight = ");
        sb.append(isValue(weight) ? wrapStringWithQuotes(convertFreeBaseWeightToEnum(weight)) : "NULL");
        sb.append(" WHERE characterID = ");
        sb.append(String.valueOf(cid));
        return sb.toString();
    }
    
    private static String removePropertyFromStatement(String tableName, int cid) {
        StringBuilder sb = new StringBuilder();
        sb.append("DELETE FROM ");
        sb.append(tableName);
        sb.append(" WHERE CID = ");
        sb.append(String.valueOf(cid));
        return sb.toString();
    }
    
    private static enum EntryStatus { New, Modified, Unmodified }
    
    private static class EntryTuple {
        private int cid;
        private FreeBaseFictionalCharacter character;
        private EntryStatus status;
        public EntryTuple(FreeBaseFictionalCharacter character, EntryStatus status, int cid) {
            this.cid = cid;
            this.character = character;
            this.status = status;
        }
        public EntryStatus getStatus() { return status; }
        public FreeBaseFictionalCharacter getCharacter() { return character; }
        public int getCid() { return cid; }
    }
    
    private static List<EntryTuple> readCharactersStatus(
            Map<Integer, FreeBaseFictionalCharacter> characters) throws SQLException {
        
        List<EntryTuple> result = new ArrayList<EntryTuple>();
        
        Connection connection = DBManager.getInstance().poolConnection();
        Statement statement = null;
        ResultSet rs = null;
        try {
            
            statement = connection.createStatement();
            
            StringBuilder sb = new StringBuilder();
            
            sb.append("SELECT characterID, freeBaseMID, newlyAdded, manuallyEdited ");
            sb.append("FROM FictionalCharacter ");
            sb.append("WHERE (");
            
            Iterator<Integer> mids = characters.keySet().iterator();
            
            sb.append("freeBaseMID = ");
            sb.append(mids.next());
            
            while (mids.hasNext()) {
                sb.append(" OR freeBaseMID = ");
                sb.append(mids.next());
            }
            
            sb.append(")");
            
            rs = statement.executeQuery(sb.toString());
            
            while (rs.next()) {
                int cid = rs.getInt(1);
                int mid = rs.getInt(2);
                boolean newlyAdded = rs.getBoolean(3);
                boolean manuallyEdited = rs.getBoolean(4);

                FreeBaseFictionalCharacter character = characters.get(mid);
                EntryStatus status = 
                    newlyAdded ? EntryStatus.New :
                    manuallyEdited ? EntryStatus.Modified : EntryStatus.Unmodified;
                
                result.add(new EntryTuple(character, status, cid));
            }
            
            return result;
            
        } finally {
            if (rs != null) {
                rs.close();
            }
            if (statement != null) {
                statement.close();
            }
            DBManager.getInstance().returnConnection(connection);
        }

    }

    private static String removeNewlyAddedMarkFromFictionalCharactersStatement() {
        return "UPDATE FictionalCharacter SET newlyAdded = 0";
    }

    // used with tables that have 2 properties
    // first property: primary key auto increment
    // second property: some "short" string value
    private static String propertyInsertStatement(String tableName, String[] values, boolean isShort) {
        
        if (!isValue(values)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT IGNORE INTO ");
        sb.append(tableName);
        sb.append(" VALUES(NULL,");
        sb.append(wrapStringWithQuotes(truncate(values[0], isShort ? Property.SHORT : Property.LONG)));
        sb.append(")");
        for (int i=1; i<values.length; i++) {
            sb.append(",(NULL,");
            sb.append(wrapStringWithQuotes(truncate(values[i], isShort ? Property.SHORT : Property.LONG)));
            sb.append(")");
        }
        return sb.toString();
    }
    
    private static String propertyInsertStatement(String tableName, String[] values) {
        return propertyInsertStatement(tableName, values, true);
    }
    
    // used with character table
    private static String characterInsertStatment(FreeBaseFictionalCharacter character) {
        
        String name = character.getName();
        int mid = midToInt(character.getMid());

        // semantically we have nothing to do with more than 1 gender so we'll pick the first one
        String[] genders = character.getGender();
        String gender = null;
        if (isValue(genders)) {
            gender = truncate(genders[0], Property.SHORT);
        }
        
        float height = character.getHeight();
        float weight = character.getWeight();
        
        String template = "INSERT IGNORE INTO FictionalCharacter(characterName,freeBaseMID,gender,height,weight) ";
        template += "VALUES({0},{1},{2},{3},{4})";

        String result = MessageFormat.format(template, 
                wrapStringWithQuotes(name),
                String.valueOf(mid),
                isValue(gender) ? wrapStringWithQuotes(gender) : "NULL", 
                isValue(height) ? wrapStringWithQuotes(convertFreeBaseHeightToEnum(height)) : "NULL", 
                isValue(weight) ? wrapStringWithQuotes(convertFreeBaseWeightToEnum(weight)) : "NULL"
            );
        
        return result;
    }
    
    // for joins between mid and name in FictionalCharacter table
    private static String insertSelectStatement(String targetTable, int mid, 
            String idProperty, String[] nameValues) {
        
        if (!isValue(nameValues)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO ");
        sb.append(targetTable);
        sb.append("(CID,");
        sb.append(idProperty);
        sb.append(")");
        sb.append("\n");
        sb.append("SELECT DISTINCT x.characterID, y.characterID");
        sb.append("\n");
        sb.append("FROM FictionalCharacter AS x,FictionalCharacter AS y");
        sb.append("\n");
        sb.append("WHERE x.freeBaseMID = ");
        sb.append(String.valueOf(mid));
        sb.append(" AND (");
        sb.append("\n");    
        
        sb.append("y.characterName = ");
        sb.append(wrapStringWithQuotes(truncate(nameValues[0], Property.NAME)));
        sb.append(" ");
        for (int i=1; i<nameValues.length; i++) {
            sb.append("OR");
            sb.append("\n");
            sb.append("y.characterName = ");
            sb.append(wrapStringWithQuotes(truncate(nameValues[i], Property.NAME)));
            sb.append(" "); 
        }

        sb.append("\n");
        sb.append(")");
        
        return sb.toString();
    }
    
    // used to insert into Chararter(Entity) tables
    private static String insertSelectStatement(String sourceTable, String targetTable, 
            int mid, String idProperty, String nameProperty, String[] nameValues) {
        
        if (!isValue(nameValues)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO ");
        sb.append(targetTable);
        sb.append("(CID,");
        sb.append(idProperty);
        sb.append(")");
        sb.append("\n");
        sb.append("SELECT DISTINCT characterID,");
        sb.append(idProperty);
        sb.append("\n");
        sb.append("FROM FictionalCharacter,");
        sb.append(sourceTable);
        sb.append("\n");
        sb.append("WHERE freeBaseMID = ");
        sb.append(String.valueOf(mid));
        sb.append(" AND (");
        sb.append("\n");    
        
        sb.append(nameProperty);
        sb.append(" = ");
        sb.append(wrapStringWithQuotes(truncate(nameValues[0],Property.SHORT)));
        sb.append(" ");
        for (int i=1; i<nameValues.length; i++) {
            sb.append("OR");
            sb.append("\n");
            sb.append(nameProperty);
            sb.append(" = ");
            sb.append(wrapStringWithQuotes(truncate(nameValues[i], Property.SHORT)));
            sb.append(" "); 
        }
        
        sb.append("\n");
        sb.append(")");
        
        return sb.toString();
    }
    
    // employment tenure methods
    private static String createTempEmploymentTenureTableStatement() {
        
        StringBuilder sb = new StringBuilder();
        
        sb.append("CREATE TEMPORARY TABLE FictionalEmploymentTenure(");
        sb.append("mid INT UNSIGNED NOT NULL, title VARCHAR(64), employer VARCHAR(64), ");
        sb.append("INDEX employerNameIndex(employer), PRIMARY KEY(mid)) ENGINE = MyISAM");
        
        return sb.toString();
    }
    
    private static String dropTempEmploymentTenureTableStatement() {
        return dropTable("FictionalEmploymentTenure");
    }
    
    private static String employmentTenureInsertStatement(int mid, String title, String employer) {
        String template = "INSERT INTO FictionalEmploymentTenure(mid,title,employer) ";
        template += "VALUES({0},{1},{2})";
        String result = MessageFormat.format(template,
            String.valueOf(mid),
            isValue(title) ? wrapStringWithQuotes(truncate(title, Property.SHORT)) : "NULL",
            wrapStringWithQuotes(truncate(employer, Property.SHORT))
        );
        return result;
    }
    
    private static String characterEmployerInsertStatement(int mid, String[] employmentMIDs) {
        
        if (!isValue(employmentMIDs)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        
        sb.append("INSERT INTO CharacterEmployer(CID,employerID,employmentTitle) ");
        sb.append("\n");
        sb.append("SELECT DISTINCT FictionalCharacter.characterId, Employer.employerID, FictionalEmploymentTenure.title");
        sb.append("\n");
        sb.append("FROM FictionalCharacter, Employer, FictionalEmploymentTenure");
        sb.append("\n");
        sb.append("WHERE FictionalEmploymentTenure.employer = Employer.employerName AND ");
        sb.append("\n");
        sb.append("freeBaseMID = ");
        sb.append(String.valueOf(mid));
        sb.append(" AND (");
        sb.append("\n");
        
        sb.append("FictionalEmploymentTenure.mid = ");
        sb.append(String.valueOf(midToInt(employmentMIDs[0])));
        for (int i=1; i<employmentMIDs.length; i++) {
            sb.append(" OR ");
            sb.append("\n");
            sb.append("FictionalEmploymentTenure.mid = ");
            sb.append(String.valueOf(midToInt(employmentMIDs[i])));
        }
        sb.append("\n");
        sb.append(")");
        
        return sb.toString();
    }
    
    // time date methods
    private static String createTempDateTimeTableStatement() {
        StringBuilder sb = new StringBuilder();
        
        sb.append("CREATE TEMPORARY TABLE FictionalDateTime(");
        sb.append("mid INT UNSIGNED NOT NULL, ");
        sb.append("standardDate ENUM('before 1900','1900-1950','1950-1959','1960-1969','1970-1979','1980-1989',");
        sb.append("'1990-1999','after 2000'),");
        sb.append("otherCalendarSystem VARCHAR(128),");
        sb.append("otherDate VARCHAR(128),");
        sb.append("PRIMARY KEY(mid)) ENGINE = MyISAM");
        
        return sb.toString();
    }
    
    private static String dropTempDateTimeTableStatement() {
        return dropTable("FictionalDateTime");
    }
    
    private static String dateTimeInsertStatement(DateTimeEntry dateTime) {
        
        String template = "INSERT INTO FictionalDateTime(mid,standardDate,otherCalendarSystem,otherDate) ";
        template += "VALUES({0},{1},{2},{3})";
        
        String result = MessageFormat.format(template,
            String.valueOf(dateTime.getMid()),
            dateTime.getStandardDate() == StandardDate.OtherDateTime ? 
                "NULL" : wrapStringWithQuotes(dateTime.getStandardDate().getName()),
            isValue(dateTime.getOtherCalendarSystem()) ? 
                    wrapStringWithQuotes(truncate(dateTime.getOtherCalendarSystem(), Property.DATE)) : "NULL",
            isValue(dateTime.getOtherDate()) ? 
                    wrapStringWithQuotes(truncate(dateTime.getOtherDate(),Property.DATE)) : "NULL"
        );

        return result;
    }
    
    private static String characterDateTimeInsertStatement(int mid, String[] dateOfBirthMIDs) {
        
        if (!isValue(dateOfBirthMIDs)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        
        sb.append("INSERT INTO CharacterDateOfBirth(CID,standardDate,otherCalendarSystem,otherDate)");
        sb.append("\n");
        sb.append("SELECT characterID,standardDate,otherCalendarSystem,otherDate");
        sb.append("\n");
        sb.append("FROM FictionalCharacter, FictionalDateTime");
        sb.append("\n");
        sb.append("WHERE freeBaseMID = ");
        sb.append(String.valueOf(mid));
        sb.append(" AND (");
        sb.append("\n");
        
        sb.append("FictionalDateTime.mid = ");
        sb.append(String.valueOf(midToInt(dateOfBirthMIDs[0])));
        for (int i=1; i<dateOfBirthMIDs.length; i++) {
            sb.append(" OR ");
            sb.append("\n");
            sb.append("FictionalDateTime.mid = ");
            sb.append(String.valueOf(midToInt(dateOfBirthMIDs[i])));           
        }
        
        sb.append("\n");
        sb.append(")");
        
        return sb.toString();
    }
    
    private static DateTimeEntry parseDateTimeLine(FreeBaseFictionalDateTime dateTime) {
        
        DateTimeEntry entry = new DateTimeEntry();
        entry.setMid(midToInt(dateTime.getMid()));

        String[] splitDate = null;
        if (isValue(dateTime.getDateTime())) {
            splitDate = dateTime.getDateTime().split("-");
        }
        
        // simple heuristic to determine best field. by far not optimal
        if (isValue(dateTime.getName()) && dateTime.getName().contains("BCE")) {
            entry.setStandardDate(StandardDate.Before1900);
        } else if (splitDate != null && splitDate.length > 0) {
            try {
                int year = Integer.parseInt(splitDate[0]);
                if (year < 1900) {
                    entry.setStandardDate(StandardDate.Before1900);
                } else if (year <= 1950) {
                    entry.setStandardDate(StandardDate.Between1900_1950);
                } else if (year <= 1959) {
                    entry.setStandardDate(StandardDate.Between1950_1959);
                } else if (year <= 1969) {
                    entry.setStandardDate(StandardDate.Between1960_1969);
                } else if (year <= 1979) {
                    entry.setStandardDate(StandardDate.Between1970_1979);
                } else if (year <= 1989) {
                    entry.setStandardDate(StandardDate.Between1980_1989);
                } else if (year <= 1999) {
                    entry.setStandardDate(StandardDate.Between1990_1999);
                } else {
                    entry.setStandardDate(StandardDate.After2000);
                }
            } catch (NumberFormatException e) {
                entry.setStandardDate(StandardDate.OtherDateTime);
            }
        } else if (isValue(dateTime.getOtherDateTime())) {
            entry.setOtherDate(dateTime.getOtherDateTime());
            if (isValue(dateTime.getOtherCalendarSystem())) {
                entry.setOtherCalendarSystem(dateTime.getOtherCalendarSystem()); 
            }
            entry.setStandardDate(StandardDate.OtherDateTime);
        } else if (isValue(dateTime.getName())) {
            entry.setOtherDate(dateTime.getName());
            entry.setStandardDate(StandardDate.OtherDateTime);
        } else {
            entry.setStandardDate(StandardDate.OtherDateTime);
        }
        
        return entry;
    }
    
    private static enum StandardDate {
        OtherDateTime("other"),
        Before1900("before 1900"),
        Between1900_1950("1900-1950"),
        Between1950_1959("1950-1959"),
        Between1960_1969("1960-1969"),
        Between1970_1979("1970-1979"),
        Between1980_1989("1980-1989"),
        Between1990_1999("1990-1999"),
        After2000("after 2000");
        
        private String name;
        private StandardDate(String name) {
            this.name = name;
        }
        public String getName() {
            return name;
        }
    }
    
    private static class DateTimeEntry {
        private int mid;
        private StandardDate standardDate;
        private String otherCalendarSystem;
        private String otherDate;
        public void setOtherCalendarSystem(String otherCalendarSystem) {
            this.otherCalendarSystem = otherCalendarSystem;
        }
        public String getOtherCalendarSystem() {
            return otherCalendarSystem;
        }
        public void setStandardDate(StandardDate standardDate) {
            this.standardDate = standardDate;
        }
        public StandardDate getStandardDate() {
            return standardDate;
        }
        public void setOtherDate(String otherDate) {
            this.otherDate = otherDate;
        }
        public String getOtherDate() {
            return otherDate;
        }
        public void setMid(int mid) {
            this.mid = mid;
        }
        public int getMid() {
            return mid;
        }
    }
    
    // marriage tables
    private static String createTempMarriedTableStatement() {
        StringBuilder sb = new StringBuilder();
        // Not temporary because we refer to it more than once in the same query
        sb.append("CREATE TABLE Marriage("); 
        sb.append("mid INT UNSIGNED, leftCol VARCHAR(128), rightCol VARCHAR(128),");
        sb.append("INDEX leftRomanticIndex(leftCol),INDEX rightRomanticIndex(rightCol),");
        sb.append("PRIMARY KEY(mid)) ENGINE = MyISAM");
        return sb.toString();
    }

    private static String dropTempMarriedTableStatement() {
        return dropTable("Marriage");
    }

    private static String marriageInsertStatement(FreeBaseGenericMediator marriage) {
        String template = "INSERT INTO Marriage(mid,leftCol,rightCol) ";
        template += "VALUES({0},{1},{2})";
        String result = MessageFormat.format(template, 
            String.valueOf(midToInt(marriage.getMid())),
            wrapStringWithQuotes(truncate(marriage.getValues()[0], Property.NAME)),
            wrapStringWithQuotes(truncate(marriage.getValues()[1], Property.NAME))
        );
        return result;
    }
    
    private static String characterMarriedToInsertStatement(int mid, String[] marriedTo) {
        
        if (!isValue(marriedTo)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        
        // we do MIN(y.characterID) because we actually only care about the name, not the person
        // so this is to efficiently point to the name string instead of storing it multiple times
        sb.append("INSERT INTO CharacterMarriedTo(CID,spouseID) ");
        sb.append("\n");
        sb.append("SELECT DISTINCT x.characterID, MIN(y.characterID) ");
        sb.append("\n");
        sb.append("FROM FictionalCharacter AS x, FictionalCharacter AS y, (");
        sb.append("(SELECT mid, leftCol, rightCol FROM Marriage) UNION ");
        sb.append("(SELECT mid, rightCol, leftCol FROM Marriage)) AS Pairs ");
        sb.append("\n");
        sb.append("WHERE x.freeBaseMID = " + mid);
        sb.append("\n");
        sb.append(" AND leftCol = x.characterName");
        sb.append(" AND rightCol = y.characterName AND (");
        sb.append("\n");

        sb.append("mid = " + String.valueOf(midToInt(marriedTo[0])));
        for (int i=1; i<marriedTo.length; i++) {
            sb.append(" OR ");
            sb.append("\n");
            sb.append("mid = " + String.valueOf(midToInt(marriedTo[i])));
        }
        
        sb.append(")");
        
        sb.append("\n");
        sb.append("GROUP BY x.characterID, y.characterName");
        
        return sb.toString();
    }
    
    // romantic tables methods
    private static String createTempRomanticTableStatement() {
        StringBuilder sb = new StringBuilder();
        // not temporary because we refer to it more than once in the same query
        sb.append("CREATE TABLE Romantic(");
        sb.append("mid INT UNSIGNED, leftCol VARCHAR(128), rightCol VARCHAR(128),");
        sb.append("INDEX leftRomanticIndex(leftCol),INDEX rightRomanticIndex(rightCol),");
        sb.append("PRIMARY KEY(mid)) ENGINE = MyISAM");
        return sb.toString();
    }
    
    private static String dropTempRomanticTableStatement() {
        return dropTable("Romantic");
    }
    
    private static String romanticInsertStatement(FreeBaseGenericMediator romantic) {
        String template = "INSERT INTO Romantic(mid,leftCol,rightCol) ";
        template += "VALUES({0},{1},{2})";
        String result = MessageFormat.format(template, 
            String.valueOf(midToInt(romantic.getMid())),
            wrapStringWithQuotes(truncate(romantic.getValues()[0], Property.NAME)),
            wrapStringWithQuotes(truncate(romantic.getValues()[1], Property.NAME))
        );
        return result;
    }
    
    private static String characterRomanticInvolvementInsertStatement(int mid, String[] romanticallyInvolvedWith) {
        
        if (!isValue(romanticallyInvolvedWith)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        
        // we do MIN(y.characterID) because we actually only care about the name, not the person
        // so this is to efficiently point to the name string instead of storing it multiple times
        sb.append("INSERT INTO CharacterRomanticallyInvolvedWith(CID,partnerID) ");
        sb.append("\n");
        sb.append("SELECT DISTINCT x.characterID, MIN(y.characterID) ");
        sb.append("\n");
        sb.append("FROM FictionalCharacter AS x, FictionalCharacter AS y, (");
        sb.append("(SELECT mid, leftCol, rightCol FROM Romantic) UNION ");
        sb.append("(SELECT mid, rightCol, leftCol FROM Romantic)) AS Pairs ");
        sb.append("\n");
        sb.append("WHERE x.freeBaseMID = " + mid);
        sb.append("\n");
        sb.append(" AND leftCol = x.characterName");
        sb.append(" AND rightCol = y.characterName AND (");
        sb.append("\n");

        sb.append("mid = " + String.valueOf(midToInt(romanticallyInvolvedWith[0])));
        for (int i=1; i<romanticallyInvolvedWith.length; i++) {
            sb.append(" OR ");
            sb.append("\n");
            sb.append("mid = " + String.valueOf(midToInt(romanticallyInvolvedWith[i])));
        }
        
        sb.append(")");
        
        sb.append("\n");
        sb.append("GROUP BY x.characterID, y.characterName");
        
        return sb.toString();
    }
    
    // siblings tables methods
    private static String createTempSibligsTableStatement() {
        StringBuilder sb = new StringBuilder();
        // Not temporary because we refer to it more than once in the same query
        sb.append("CREATE TABLE Siblings("); 
        sb.append("siblingID INT UNSIGNED AUTO_INCREMENT, mid INT UNSIGNED, ");
        sb.append("leftCol VARCHAR(128), rightCol VARCHAR(128), ");
        sb.append("INDEX midSiblingsIndex(mid),");
        sb.append("INDEX leftSiblingsIndex(leftCol),INDEX rightSiblingsIndex(rightCol),");
        sb.append("PRIMARY KEY(siblingID)) ENGINE = MyISAM");
        return sb.toString();
    }

    private static String dropTempSibligsTableStatement() {
        return dropTable("Siblings");
    }

    private static String siblingsInsertStatement(FreeBaseGenericMediator siblings) {

        List<Pair<String,String>> siblingPairs = new ArrayList<Pair<String,String>>();
        for (int i=0; i<siblings.getValues().length; i++) {
            for (int j=i+1; j<siblings.getValues().length; j++) {
                siblingPairs.add(new Pair<String,String>(
                    truncate(siblings.getValues()[i],Property.NAME), 
                    truncate(siblings.getValues()[j],Property.NAME)
                ));
            }
        }
        
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO Siblings(mid,leftCol,rightCol) VALUES");
        
        String template = "({0},{1},{2}),";
        for (Pair<String,String> pair : siblingPairs) {
            String result = MessageFormat.format(template, 
                String.valueOf(midToInt(siblings.getMid())),
                wrapStringWithQuotes(pair.getLeft()),
                wrapStringWithQuotes(pair.getRight())
            );
            sb.append(result);
        }
        
        sb.setLength(sb.length() - 1);
        
        return sb.toString();
        
    }
    
    private static String characterSiblingsInsertStatement(int mid, String[] siblings) {
        
        if (!isValue(siblings)) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        
        // we do MIN(y.characterID) because we actually only care about the name, not the person
        // so this is to efficiently point to the name string instead of storing it multiple times
        sb.append("INSERT INTO CharacterSibling(CID,siblingID) ");
        sb.append("\n");
        sb.append("SELECT DISTINCT x.characterID, MIN(y.characterID) ");
        sb.append("\n");
        sb.append("FROM FictionalCharacter AS x, FictionalCharacter AS y, (");
        sb.append("(SELECT mid, leftCol, rightCol FROM Siblings) UNION ");
        sb.append("(SELECT mid, rightCol, leftCol FROM Siblings)) AS Pairs ");
        sb.append("\n");
        sb.append("WHERE x.freeBaseMID = " + mid);
        sb.append("\n");
        sb.append(" AND leftCol = x.characterName");
        sb.append(" AND rightCol = y.characterName AND (");
        sb.append("\n");

        sb.append("mid = " + String.valueOf(midToInt(siblings[0])));
        for (int i=1; i<siblings.length; i++) {
            sb.append(" OR ");
            sb.append("\n");
            sb.append("mid = " + String.valueOf(midToInt(siblings[i])));
        }
        
        sb.append(")");
        
        sb.append("\n");
        sb.append("GROUP BY x.characterID, y.characterName");
        
        return sb.toString();
    }
    
    // general utils
    private static String truncate(String str, Property prop) {
        return str == null || str.length() <= prop.getMaxSize() ? str : str.substring(0, prop.getMaxSize());
    }
    
    private static String dropTable(String tableName) {
        return "DROP TABLE " + tableName;
    }
    
    private static File loadFile(File freeBaseRoot, String filename) throws FileNotFoundException {
        File file = new File(freeBaseRoot, filename + ".tsv");
        if (!file.isFile()) {
            throw new FileNotFoundException("Failed finding " + filename + ".tsv in freebase root directory");
        }
        return file;
    }
    
    // converters
    private static String convertFreeBaseHeightToEnum(float height) {
        if (height < 1.5) {
            return "below 1.5m";
        } else if (height <= 1.75) {
            return "1.5-1.75m";
        } else if (height <= 2) {
            return "1.75-2m";
        } else {
            return "above 2m";
        }
    }
    
    private static String convertFreeBaseWeightToEnum(float weight) {
        if (weight < 50) {
            return "below 50kg";
        } else if (weight <= 75) {
            return "50-75";
        } else if (weight <= 100) {
            return "75-100";
        } else {
            return "above 100kg";
        }
    }
    
    private static class BatchExecutor {
        
        private int batchCount = 0;
        private Connection connection;
        private Statement statement;
        boolean previousAutoCommit;
        
        public BatchExecutor() throws SQLException {
            connection = DBManager.getInstance().poolConnection();
            previousAutoCommit = connection.getAutoCommit();
            connection.setAutoCommit(false);
            statement = connection.createStatement();
        }
        
        private void execute(boolean force) throws SQLException {
            int threshold = force ? 1 : BATCH_SIZE;
            if (batchCount >= threshold) {
                statement.executeBatch();
                connection.commit();
                batchCount = 0;
            }
        }
        
        public void forceBatch() throws SQLException {
            execute(true);
        }
        
        public void addBatch(String sql) throws SQLException {
            if (sql == null) {
                return;
            }
            statement.addBatch(sql);
            batchCount += 1;
            execute(false);
        }
        
        public void close() throws SQLException {
            try {
                if (statement != null) {
                    statement.close();
                }
                if (connection != null) {
                    connection.setAutoCommit(previousAutoCommit);
                }
            } finally {
                if (connection != null) {
                    DBManager.getInstance().returnConnection(connection);
                }
            }
        } 
    }
    
    private static enum Property {
        SHORT(64), LONG(1024), DATE(128), NAME(128);
        int maxSize;
        private Property(int maxSize) {
            this.maxSize = maxSize;
        }
        public int getMaxSize() {
            return maxSize;
        }
    }
    
}
